package com.lefu.risk.storm.rule;

import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import com.lefu.risk.storm.common.constants.Constant;
import com.lefu.risk.storm.utils.JedisPoolUtil;

public class OrderAmountBolt extends BaseRichBolt{

	private static final long serialVersionUID = 1L;
	
	private OutputCollector collector;

	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
	}

	@Override
	public void execute(Tuple input) {
		JedisPoolUtil redisUtil = null;
		try {
			redisUtil = new JedisPoolUtil(Constant.REDIS_PROPERTIES_NAME);
			String word = input.getStringByField("word");
			Double amount = input.getDoubleByField("amount");
			String maxAmount = redisUtil.get(word+"_max_amount");
			String minAmount = redisUtil.get(word+"_min_amount");
			
			Double _maxAmount = null == maxAmount ? 0.00: Double.valueOf(maxAmount);
			Double _minAmount = null == minAmount ? amount: Double.valueOf(minAmount);
			
			redisUtil.hincrByFloat("total_amount",word+"_total_amount", amount);
			
			if(amount > _maxAmount){
				redisUtil.set(word+"_max_amount", String.format("%.2f", amount));
			}
			if(amount <= _minAmount){
				redisUtil.set(word+"_min_amount", String.format("%.2f", amount));
			}
			collector.ack(input);
			System.err.println(word);
		} catch (Exception e) {
			e.printStackTrace();
			collector.fail(input);
		}finally{
			if (null !=redisUtil )redisUtil.close();
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		
	}

}
